#include "FIFOQueue.h"
#include <atomic>
#include <cassert>
#include <cstdint>
#include <cstdio>
#include <cstdlib>
#include <cstring>
#include <memory>
#include <sstream>
#include <unistd.h>
#include "PlatLogger.h"
#include "Util.h"
#include "XnccEventfd.h"
#include "XnccType.h"
namespace xncc {
namespace msg {
    using namespace xncc::types;

    // #define aligned_queuelen_t __attribute__((aligned(64))) queue_size_t

#define aligned_queuelen_t queue_size_t

    class FIFOQueueElement
    {
      public:
        FIFOQueueElement* next = nullptr;
        explicit FIFOQueueElement(queue_size_t size) : MASK(size - 1)
        {
            if (!shared::CommonBusi::is_power_of_2(size)) {
                throw std::logic_error("size must be power of 2");
            }
            in           = 0;
            out          = 0;
            currReadLen  = 0;
            currWriteLen = 0;
            data         = reinterpret_cast<char*>(malloc(size));
            assert(data != nullptr);
        }
        ~FIFOQueueElement()
        {
            free(data);
            data = nullptr;
            TRACE_INFO << "free FIFOQueueElement"
                       << " " << dump();
        }

        [[nodiscard]] queue_size_t capacity() const { return MASK + static_cast<queue_size_t>(1); }

        char* getWriteBuffer(queue_size_t len)
        {
            const auto cached_out  = out;
            const auto cached_in   = in;
            const auto read_index  = static_cast<queue_size_t>(cached_out & MASK);
            const auto write_index = static_cast<queue_size_t>(cached_in & MASK);
            const auto unused      = (capacity() - static_cast<queue_size_t>(cached_in - cached_out));
            // case1
            //                              w_index        r_index      Final
            //|-------------------------------|-------------|------------|---------------|
            //                                |<-  unused ->|            |               |
            // 如果写数据不超过unused 不影响Final 写数据超过unused 写入失败 不影响Final

            // case2
            //                              w_index        Final      r_index
            //|-------------------------------|-------------|------------|---------------|
            //                                |<-         unused       ->|               |
            // 这种情况不会出现 rindex 遇到final 就会回到起点

            // case3
            //                              r_index        w_index      Final
            //|-------------------------------|-------------|------------|---------------|
            //|<-             unused        ->|             |<-         unused         ->|
            // 右边的unused的已经read了 或者没有read（只发生在刚开始写数据的时候 Final是size）已经read 此次write
            // 会影响Final
            if (len > unused) {
                return nullptr;
            }
            if (static_cast<queue_size_t>(capacity() - write_index) >= len) {
                currWriteLen = len;
                if (static_cast<queue_size_t>(write_index + len) > currentFinal) {
                    currentFinal = capacity();
                }
                return data + write_index;
            }
            if (len <= read_index) {
                currWriteLen = len + static_cast<queue_size_t>(capacity() - write_index);
                currentFinal = write_index;
                return data;
            }
            return nullptr;
        }

        void finishWrite()
        {
            smp_wmb();
            const auto cached_in = in;
            const auto result    = static_cast<queue_size_t>(cached_in + currWriteLen);
            in                   = result;
        }

        char* getReadBuffer(queue_size_t len)
        {
            const auto cached_in   = in;
            const auto cached_out  = out;
            const auto read_index  = static_cast<queue_size_t>(cached_out & MASK);
            const auto write_index = static_cast<queue_size_t>(cached_in & MASK);
            if (static_cast<queue_size_t>(cached_in - cached_out) >= len) {
                __asm__ __volatile__("" ::: "memory");  // prevent new write_index old final
                if (read_index == currentFinal) {
                    if (write_index >= len) {
                        currReadLen = len + static_cast<queue_size_t>(capacity() - read_index);
                        return data;
                    }
                    fprintf(stderr, "read  what happened ? %s \n", dump().c_str());
                }
                else {
                    currReadLen = len;
                    return data + read_index;
                }
            }
            else {
                return nullptr;
            }
            return nullptr;
        }

        void finishRead()
        {
            smp_rmb();
            const auto cached_out = out;
            const auto result     = static_cast<queue_size_t>(cached_out + currReadLen);
            out                   = result;
        }

        [[nodiscard]] std::string dump() const
        {
            std::ostringstream oss;
            oss << "in:" << in << " out:" << out << " in_off:" << (in & MASK) << " out_off:" << (out & MASK)
                << " size:" << (capacity() / 1024 / 1024) << "MB " << (capacity() / 1024)
                << "KB currReadLen:" << currReadLen << " currWriteLen:" << currWriteLen;
            return oss.str();
        }

        //   private:
        //     inline queue_size_t unused()
        //     {
        //         return static_cast<queue_size_t>((MASK + queue_size_t(1)) - static_cast<queue_size_t>(in - out));
        //     }

      private:
        const aligned_queuelen_t    MASK;
        volatile aligned_queuelen_t out;
        volatile aligned_queuelen_t currReadLen;
        volatile aligned_queuelen_t in;
        volatile aligned_queuelen_t currWriteLen;
        volatile aligned_queuelen_t currentFinal{};

        char* data;
    };

    static std::atomic<types::queue_id_t> INIT_QUEUE_ID(1);
    FIFOQueue::FIFOQueue(queue_size_t size, bool autoExpand)
        : autoExpand_(autoExpand), lastReadIndex_(0), isNeedNotify_(false), queueId_(INIT_QUEUE_ID.fetch_add(1))
    {
        auto* ptr = new FIFOQueueElement(size);
        readPtr_  = ptr;
        writePtr_ = ptr;
    }
    void FIFOQueue::setNeedNotify()
    {
        notifyfd_     = (std::make_shared<sys::EventFd>());
        isNeedNotify_ = true;
    }

    std::string FIFOQueue::dump() const
    {
        std::stringstream ss;
        ss << "readptr:" << readPtr_->dump() << "  "
           << " writeptr:" << writePtr_->dump();

        return ss.str();
    }

    // void FIFOQueue::updateCurrentWritePtr()
    // {
    //     // 写指针的改变 以及使用都是在写线程操作，因此不存在多线程竞争问题
    //     std::lock_guard<std::mutex> lk(mutex_);
    //     int                         maxIndex = bufferList_.rbegin()->first;
    //     queue_size_t                oldCap   = bufferList_.rbegin()->second->capacity();
    //     queue_size_t                newcap   = oldCap * 2;
    //     if (oldCap == types::queue_size_t_limit<queue_size_t>::max()) {
    //         newcap = oldCap;
    //     }
    //     maxIndex++;
    //     auto ptr              = std::make_shared<FIFOQueueElement>(newcap);  // 扩容
    //     bufferList_[maxIndex] = ptr;
    //     writePtr_             = ptr;
    // }

    // void FIFOQueue::updateCurrentReadPtr()
    // {
    //     // 读指针的改变 以及使用都是在读线程操作，因此不存在多线程竞争问题
    //     std::lock_guard<std::mutex> lk(mutex_);
    //     int                         originIndex = lastReadIndex_;
    //     lastReadIndex_++;
    //     auto iter = bufferList_.find(lastReadIndex_);
    //     if (iter != bufferList_.end()) {
    //         readPtr_ = iter->second;
    //         bufferList_.erase(originIndex);
    //     }
    //     else {
    //         TRACE_ERROR << "what happened " << bufferList_.size() << " " << originIndex << " " << lastReadIndex_
    //                     << " dump read buffer " << readPtr_->dump();
    //         lastReadIndex_--;
    //     }
    // }

    char* FIFOQueue::tryGetWriteBuffer(queue_size_t len) { return writePtr_->getWriteBuffer(len); }

    char* FIFOQueue::getWriteBuffer(queue_size_t len)
    {
        char* buf = writePtr_->getWriteBuffer(len);
        if (buf == nullptr) {  // buffer full failed
            if (autoExpand_) {
                // updateCurrentWritePtr();               // expand
                // allocte new node
                const queue_size_t oldCap = writePtr_->capacity();
                queue_size_t       newcap = oldCap * 2;
                if (oldCap == types::queue_size_t_limit<queue_size_t>::max()) {
                    newcap = oldCap;
                }
                auto* ptr       = new FIFOQueueElement(newcap);
                writePtr_->next = ptr;
                writePtr_       = ptr;
                buf             = writePtr_->getWriteBuffer(len);  // 只扩容一次

                // del old node //注意如果读取的慢的话 当不再重新分配内存之后 可能有残留内存没有被释放
                // 因为触发释放得新分配内存 //如果每次都检测 导致性能下降
                auto* tmp_del = delList_;
                delList_      = nullptr;
                while (tmp_del) {
                    auto del = tmp_del;
                    tmp_del  = tmp_del->next;
                    delete del;
                }
            }
            else {
                while ((buf = writePtr_->getWriteBuffer(len)) == nullptr) {}
            }
        }

        return buf;
    }
    void FIFOQueue::finishWrite()
    {
        writePtr_->finishWrite();
        if (isNeedNotify_) {
            uint64_t one = 1;
            auto     n   = notifyfd_->write(one);
            if (n != sizeof one) {
                SYSTEM_ERROR << "event write failed fd:"_s << notifyfd_->Fd() << " queue id:"_s << queueId_;
            }
        }
    }

    void FIFOQueue::finishWriteNotNotify() { writePtr_->finishWrite(); }

    char* FIFOQueue::getReadBuffer(queue_size_t len)
    {
        char* buf = readPtr_->getReadBuffer(len);
        if (autoExpand_) {
            if (buf == nullptr) {
                if (readPtr_->next != nullptr) {  // 有新的buffer存在
                    buf = readPtr_->getReadBuffer(len);
                    if (buf != nullptr) {
                        return buf;
                    }
                    // switch  读的慢写的快 才导致新分配节点 因为节点的内存释放 不应该放到读的部分
                    auto* ptr      = readPtr_->next;
                    readPtr_->next = delList_;  // 当前节点挂载到dellist的头部
                    delList_       = readPtr_;
                    readPtr_       = ptr;
                    buf            = readPtr_->getReadBuffer(len);
                }

                // bool needUpdate = true;
                // if (!isNeedNotify_) {  // 不需要通知就进行读操作的read
                //     // 可能没有数据的时候就开始读了，这个时候自动扩容的情况下 没有扩容情况下 不更新读指针
                //     std::size_t listSize;
                //     {
                //         std::lock_guard<std::mutex> lk(mutex_);
                //         listSize = bufferList_.size();
                //     }
                //     if (listSize == 1) {
                //         needUpdate = false;
                //     }
                // }
                // if (needUpdate) {  // 两种情况  一种新buffer 还没有数据 因为读的快 写还有做finish  另一种读发现是空
                //                    // 但是旧buffer还有数据 恰好被写入了
                //     buf = readPtr_->getReadBuffer(len);
                //     if (buf) {
                //         return buf;
                //     }
                //     else {
                //         updateCurrentReadPtr();
                //         buf = readPtr_->getReadBuffer(len);
                //     }
                // }
            }
        }
        return buf;
    }
    void FIFOQueue::finishRead() { readPtr_->finishRead(); }
}  // namespace msg
}  // namespace xncc